package com.zwh.rocketmq;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class MQProducerUtils {

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;

    // 建议正常规模项目统一用一个TOPIC
    private static final String topic = "RLT_TEST_TOPIC";

    // 直接注入使用，用于发送消息到broker服务器
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 同步发送数据
     * @param topic
     * @param tags
     * @param msgBody object对象形式发送
     * @param keys
     * @return
     */
    public SendResult sendMsg(String topic, String tags, Object msgBody, String keys) {
        String destination = topic;
        if (StringUtils.isNotBlank(tags)) {
            // 带tag的消息，直接在topic后面加上":tag"， 底层会进行自动转换
            destination = destination + ":" + tags;
        }
        /*
         headers中的key：
            id: 表示消息的id,和mq控制台上的messageId不一样，没什么作用
            TAGS：消息标签
            KEYS： 表示消息的key，用于哈希索引，查找使用
         */
        Message<Object> message = MessageBuilder.withPayload(msgBody)
                .setHeader(RocketMQHeaders.KEYS, keys)
                .build();
        // 同步发送
        /**
         * 默认的converter：new RocketMQMessageConverter.getMessageConverter()
         * 默认情况下object会序列化成json数据格式的byte数据（utf-8编码），在控制台是以json字符串展示了
         * 如果object对象要序列化成指定格式的内容：
         *  可通过这个指定：
         *  header.put("contentType", "application/json")
         *  header.put("contentType", "application/octet-stream")
         *
         *  消费者可以接收String类型数据，或者序列化时的Object类型的数据（内部会做类似反序列化操作JSON.parse("messageBody", Type)）
         */
        return rocketMQTemplate.syncSend(destination, message);
    }

    /**
     * 发送消息
     * @param topic 主题
     * @param tags 消息标签，用于消费者过滤消息
     * @param msgBody 消息内容
     * @param keys 消息的key，默认会使用uuid生成，用于消息队列中查询
     * @return sendResult为返回的发送结果
     */
    public SendResult sendMsg(String topic, String tags, String msgBody, String keys) {
        String destination = topic;
        if (StringUtils.isNotBlank(tags)) {
            destination = destination + ":" + tags;
        }
        Message<String> message = MessageBuilder.withPayload(msgBody)
                .setHeader(RocketMQHeaders.KEYS, keys)
                .build();
        // 同步发送
        return rocketMQTemplate.syncSend(destination, message);
    }

    /**
     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
     * （适合对响应时间敏感的业务场景）
     */
    public void sendAsyncMsg(String topic, String tags, String msgBody, String keys, SendCallback callback) {
        String destination = topic;
        if (StringUtils.isNotBlank(tags)) {
            destination = destination + ":" + tags;
        }
        Message<String> message = MessageBuilder.withPayload(msgBody)
                .setHeader(RocketMQHeaders.KEYS, keys)
                .build();
        rocketMQTemplate.asyncSend(destination, message, callback);

//        new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//                // 处理消息发送成功逻辑
//            }
//            @Override
//            public void onException(Throwable throwable) {
//                // 处理消息发送异常逻辑
//            }
//        }
    }

    /**
     * 发送一次，不管有没有ack响应，类似udp协议
     * @param topic
     * @param tags
     * @param msgBody
     * @param keys
     */
    public void sendOneWay(String topic, String tags, String msgBody, String keys) {
        String destination = topic;
        if (StringUtils.isNotBlank(tags)) {
            destination = destination + ":" + tags;
        }
        Message<String> message = MessageBuilder.withPayload(msgBody)
                .setHeader(RocketMQHeaders.KEYS, keys)
                .build();
        rocketMQTemplate.sendOneWay(destination, message);
    }

    /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public void sendDelayMsg(String msgBody, int delayLevel) {
        rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build(), messageTimeOut, delayLevel);
    }

    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果，如日志）
     */
    public void sendOneWayMsg(String msgBody) {
        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(msgBody).build());
    }

    /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     */
    public SendResult sendTagMsg(String msgBody) {
        return rocketMQTemplate.syncSend(topic + ":tag2", MessageBuilder.withPayload(msgBody).build());
    }

}
